#!/usr/bin/env python
# -*- coding: utf-8 -*-

from __future__ import with_statement

from cStringIO import StringIO
import itertools
import unittest

from ws9 import stream, line


original_data = """The Zen of Python, by Tim Peters
Beautiful is better than ugly.
Explicit is better than implicit.
Simple is better than complex.
Complex is better than complicated.
Flat is better than nested.
"""
split_data = ["The Zen of Python, by ", "Tim Peters\nBeautiful",
    " is better than ugly.\n", "Explicit is better",
    " than implicit.\nSimple", " is better than complex.\nComplex is",
    " better than complicated.\nFlat ", "is better than nested.\n"]
split_mixed_data = ["The Zen of Python, by ", "Tim Peters\nBeautiful",
    " is better than ugly.\r", "Explicit is better",
    " than implicit.\r\nSimple", " is better than complex.\nComplex is",
    " better than complicated.\rFlat ", "is better than nested.\r\n"]
split_crlf_data = [s.replace('\n', '\r\n') for s in split_data]
cr_lines = [l + '\r' for l in original_data.splitlines()]
cr_data = ''.join(cr_lines)
indented_data = """The Zen of Python, by Tim Peters
    Beautiful is better than ugly.
    Explicit is better than implicit.
\t    Simple is better than complex.
\t\tComplex is better than complicated.
\t    \tFlat is better than nested.
"""


class TestIteratorStream(unittest.TestCase):
    
    def setUp(self):
        self.iterator = iter(split_data)
        self.istream = stream.IteratorStream(self.iterator)
        self.sstream = StringIO(original_data)
    
    def tearDown(self):
        close(self.istream, self.sstream)
    
    def testRead(self):
        value = []
        curr_st_data = self.istream.read(15)
        while curr_st_data:
            value.append(curr_st_data)
            curr_st_data = self.istream.read(15)
        comparison = []
        curr_ss_data = self.sstream.read(15)
        while curr_ss_data:
            comparison.append(curr_ss_data)
            curr_ss_data = self.sstream.read(15)
        self.assertEqual(value, comparison)
    
    def testReadline(self):
        value = []
        curr_line = self.istream.readline()
        while curr_line:
            value.append(curr_line)
            curr_line = self.istream.readline()
        comparison = self.sstream.readlines()
        self.assertEqual(value, comparison)
    
    def testReadlines(self):
        value = self.istream.readlines()
        comparison = self.sstream.readlines()
        self.assertEqual(value, comparison)
    
    def testMixedReadReadline(self):
        value = []
        comparison = []
        curr_val = self.istream.read(10)
        curr_comp = self.sstream.read(10)
        i = 0
        while curr_val and curr_comp:
            value.append(curr_val)
            comparison.append(curr_comp)
            if i % 2:
                curr_val = self.istream.read(10)
                curr_comp = self.sstream.read(10)
            else:
                curr_val = self.istream.readline()
                curr_comp = self.sstream.readline()
        self.assertEqual(value, comparison)
    
    def testContextMgmtEntry(self):
        with stream.IteratorStream(iter(split_data)) as istream:
            read_data = istream.read()
        self.assertEqual(read_data, original_data)
    
    def testContextMgmtCleanExit(self):
        with self.istream:
            pass
        self.assert_(self.istream.closed,
            "IteratorStream does not close on clean context management exit.")
    
    def testContextMgmtDirtyExit(self):
        def dirty_contxtmgr_exit(istream):
            with istream:
                raise Exception('This exception should be noticed.')
        self.assertRaises(Exception, dirty_contxtmgr_exit, self.istream)
    
    def testIteration(self):
        for iline, sline in itertools.izip(self.istream, self.sstream):
            self.assertEqual(iline, sline)
    
    def testNoNewlines(self):
        data = self.istream.read()
        self.assertRaises(AttributeError, getattr, self.istream, 'newlines')
    
    def testClosedRead(self):
        self.istream.close()
        self.assertRaises(ValueError, self.istream.read)
    
    def testClosedReadline(self):
        self.istream.close()
        self.assertRaises(ValueError, self.istream.readline)
    
    def testClosedReadlines(self):
        self.istream.close()
        self.assertRaises(ValueError, self.istream.readlines)
    
    def testClosedNext(self):
        self.istream.close()
        self.assertRaises(StopIteration, self.istream.next)


class TestUniversalIteratorStream(unittest.TestCase):
    
    def setUp(self):
        self.iterator = iter(split_crlf_data)
        self.istream = stream.IteratorStream(self.iterator, mode='U')
        self.sstream = StringIO(original_data)
    
    def tearDown(self):
        close(self.istream, self.sstream)
    
    def testRead(self):
        value = []
        curr_st_data = self.istream.read(15)
        while curr_st_data:
            value.append(curr_st_data)
            curr_st_data = self.istream.read(15)
        comparison = []
        curr_ss_data = self.sstream.read(15)
        while curr_ss_data:
            comparison.append(curr_ss_data)
            curr_ss_data = self.sstream.read(15)
        self.assertEqual(value, comparison)
    
    def testReadline(self):
        value = []
        curr_line = self.istream.readline()
        while curr_line:
            value.append(curr_line)
            curr_line = self.istream.readline()
        comparison = self.sstream.readlines()
        self.assertEqual(value, comparison)
    
    def testReadlines(self):
        value = self.istream.readlines()
        comparison = self.sstream.readlines()
        self.assertEqual(value, comparison)
    
    def testMixedReadReadline(self):
        value = []
        comparison = []
        curr_val = self.istream.read(10)
        curr_comp = self.sstream.read(10)
        i = 0
        while curr_val and curr_comp:
            value.append(curr_val)
            comparison.append(curr_comp)
            if i % 2:
                curr_val = self.istream.read(10)
                curr_comp = self.sstream.read(10)
            else:
                curr_val = self.istream.readline()
                curr_comp = self.sstream.readline()
        self.assertEqual(value, comparison)
    
    def testIteration(self):
        for iline, sline in itertools.izip(self.istream, self.sstream):
            self.assertEqual(iline, sline)
    
    def testNewlinesSingle(self):
        data = self.istream.read()
        self.assertEqual(self.istream.newlines, '\r\n')
    
    def testNewlinesMixed(self):
        mixed_istream = stream.IteratorStream(iter(split_mixed_data), mode='U')
        data = mixed_istream.read()
        self.assertEqual(mixed_istream.newlines, ('\n', '\r', '\r\n'))
        mixed_istream.close()


class TestIteratorStreamFullBuffer(TestIteratorStream):
    
    def setUp(self):
        self.iterator = iter(split_data)
        self.istream = stream.IteratorStream(self.iterator, full_buffer=True)
        self.sstream = StringIO(original_data)
    
    def testContextMgmtEntry(self):
        with stream.IteratorStream(
            iter(split_data), full_buffer=True) as istream:
            read_data = istream.read()
        self.assertEqual(read_data, original_data)


class TestUniversalIteratorStreamFullBuffer(TestUniversalIteratorStream):
    
    def setUp(self):
        self.iterator = iter(split_crlf_data)
        self.istream = stream.IteratorStream(self.iterator, mode='U',
            full_buffer=True)
        self.sstream = StringIO(original_data)
    
    def testNewlinesMixed(self):
        mixed_istream = stream.IteratorStream(iter(split_mixed_data), mode='U',
            full_buffer=True)
        data = mixed_istream.read()
        self.assertEqual(mixed_istream.newlines, ('\n', '\r', '\r\n'))
        mixed_istream.close()


class TestLineReader(unittest.TestCase):
    
    def setUp(self):
        self.sstream = StringIO(cr_data)
        self.lstream = stream.line_reader(self.sstream)
        self.lines = iter(cr_lines)
    
    def tearDown(self):
        self.sstream.close()
    
    def testLine(self):
        for lr_line, it_line in itertools.izip(self.lstream, self.lines):
            self.assertEqual(lr_line, it_line)
    
    def testJoinLines(self):
        self.assertEqual(''.join(self.lstream), self.sstream.getvalue())


class TestSplitFirstLine(unittest.TestCase):
    
    def setUp(self):
        self.lf = """First Line.
Second Line.
Third Line."""
        self.cr = self.lf.replace('\n', '\r')
        self.crlf = self.lf.replace('\n', '\r\n')
        self.first_lf = 'First Line.\n'
        self.rest_lf = 'Second Line.\nThird Line.'
        self.first_cr = 'First Line.\r'
        self.rest_cr = 'Second Line.\rThird Line.'
        self.first_crlf = 'First Line.\r\n'
        self.rest_crlf = 'Second Line.\r\nThird Line.'
    
    def testLF(self):
        first, rest = stream.split_first_line(self.lf)
        self.assertEqual(first, self.first_lf)
        self.assertEqual(rest, self.rest_lf)
    
    def testCR(self):
        first, rest = stream.split_first_line(self.cr)
        self.assertEqual(first, self.first_cr)
        self.assertEqual(rest, self.rest_cr)

    def testCRLF(self):
        first, rest = stream.split_first_line(self.crlf)
        self.assertEqual(first, self.first_crlf)
        self.assertEqual(rest, self.rest_crlf)


class TestLineOperation(unittest.TestCase):
    
    def setUp(self):
        self.line = line.Line('    Hello, World!\r\n')
    
    def testLineDelEnding(self):
        line_op = stream.LineOperation('DELETE', attribute='ending')
        self.assertEqual(line_op(self.line), '    Hello, World!')
    
    def testLineSetEnding(self):
        line_op = stream.LineOperation('SET', attribute='ending', value='\n')
        self.assertEqual(line_op(self.line), '    Hello, World!\n')
    
    def testLineDelIndent(self):
        line_op = stream.LineOperation('DELETE', attribute='indent')
        self.assertEqual(line_op(self.line), 'Hello, World!\r\n')
    
    def testLineSetIndent(self):
        line_op = stream.LineOperation('SET', attribute='indent', value='\t')
        self.assertEqual(line_op(self.line), '\tHello, World!\r\n')
    
    def testLineNoneEnding(self):
        line_op = stream.LineOperation('NONE', attribute='ending')
        self.assertEqual(line_op(self.line), self.line)
    
    def testLineNoneIndent(self):
        line_op = stream.LineOperation('NONE', attribute='indent')
        self.assertEqual(line_op(self.line), self.line)
    
    def testSetOperation(self):
        self.assertRaises(ValueError, stream.LineOperation, 'ILLEGAL_OP')
    
    def testEquality(self):
        line_op1 = stream.LineOperation('SET', attribute='indent', value='\t')
        line_op2 = stream.LineOperation('SET', attribute='indent', value='\t')
        line_op3 = stream.LineOperation('SET', attribute='ending', value='\n')
        self.assertEqual(line_op1, line_op2)
        self.assertNotEqual(line_op1, line_op3)
        self.assertNotEqual(line_op2, line_op3)
    
    def testValueAttribute(self):
        set_op = stream.LineOperation('SET', attribute='indent', value='\t')
        del_op = stream.LineOperation('DELETE', attribute='indent')
        self.assert_(hasattr(set_op, 'value'))
        self.failIf(hasattr(del_op, 'value'))


class TestStreamDocument(unittest.TestCase):
    
    def setUp(self):
        self.stream = StringIO(indented_data)
        self.stream2 = StringIO(indented_data)
        self.doc = stream.StreamDocument(self.stream2)
    
    def tearDown(self):
        close(self.stream, self.stream2)
    
    def testIterator(self):
        for doc_line in self.doc:
            self.assertEqual(doc_line, self.stream.readline())
    
    def testGetEndings(self):
        for doc_line in self.doc:
            pass
        self.assertEqual(self.doc.endings, '\n')
    
    def testSetEndings(self):
        self.doc.endings = '\r'
        for doc_line in self.doc:
            self.assertEqual(doc_line,
                self.stream.readline().replace('\n', '\r'))
    
    def testDelEndings(self):
        del self.doc.endings
        for doc_line in self.doc:
            self.assertEqual(doc_line, self.stream.readline().rstrip('\r\n'))
    
    def testGetIndents(self):
        for doc_line in self.doc:
            pass
        self.assertEqual(self.doc.indent_type, frozenset(['\t', ' ', '\t ']))
    
    def testSetIndents(self):
        self.doc.indents = ' '
        doc_list = list(iter(self.doc))
        self.assert_(doc_list[3].indent, ' ' * 8)
        self.assert_(doc_list[4].indent, ' ' * 8)
        self.assert_(doc_list[5].indent, ' ' * 12)
    
    def testDelIndents(self):
        del self.doc.indent_type
        doc_string = ''.join(map(str, iter(self.doc)))
        self.assertEqual(indented_data.replace('\t', '    '), doc_string)
    
    def testIterMode(self):
        self.doc = iter(self.doc)
        self.assertRaises(ValueError, setattr, self.doc, 'endings', '\r\n')
        self.assertRaises(ValueError, setattr, self.doc, 'indent_type', '\t')
    
    def testNotIterMode(self):
        self.assert_(self.doc.endings is None)
        self.assert_(self.doc.indent_type is None)


def close(*args):
    for arg in args:
        arg.close()

if __name__ == '__main__':
    unittest.main()
